package com.roy.bloom;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;

import com.google.common.hash.Funnels;
import com.google.common.hash.Hashing;
import com.google.common.primitives.Longs;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;

public class MyBloomFilter {
	// 预计插入量
	private long expectedInsertions = 1000;
	// 可接受的错误率
	private double fpp = 0.001;

	private final int numHashFunctions;

	private BitArray bits = null;

	private static final String DEFAULT_INDEX = "default";
	
	private JedisCluster jedis = null;

	public MyBloomFilter() {
		this.expectedInsertions = 1000;
		this.fpp = 0.001;

		// bit数组长度
		long numBits = optimalNumOfBits(expectedInsertions, fpp);
		// hash函数数量
		this.numHashFunctions = optimalNumOfHashFunctions(expectedInsertions,
				numBits);
		this.bits = new BitArray(numBits);
	}

	public MyBloomFilter(long expectedInsertions, double fpp,JedisCluster jedis) {
		this.expectedInsertions = expectedInsertions;
		this.fpp = fpp;

		// bit数组长度
		long numBits = optimalNumOfBits(expectedInsertions, fpp);
		// hash函数数量
		this.numHashFunctions = optimalNumOfHashFunctions(expectedInsertions,
				numBits);
		this.bits = new BitArray(numBits);
		this.jedis = jedis;
	}
	
	public BitArray getBitArray() {
		return this.bits;
	}

	// 计算hash函数个数 方法来自guava
	private int optimalNumOfHashFunctions(long n, long m) {
		return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
	}

	// 计算bit数组长度 方法来自guava
	private long optimalNumOfBits(long n, double p) {
		if (p == 0) {
			p = Double.MIN_VALUE;
		}
		return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
	}

	/**
	 * 判断keys是否存在于集合where中
	 * @throws Exception 
	 */
	public boolean mightContain(String key) throws Exception {
		long bitSize = bits.bitSize();
		byte[] bytes = Hashing.murmur3_128()
				.hashObject(key, Funnels.stringFunnel(StandardCharsets.UTF_8))
				.asBytes();
		long hash1 = lowerEight(bytes);
		long hash2 = upperEight(bytes);

		long combinedHash = hash1;
		for (int i = 0; i < numHashFunctions; i++) {
			// Make the combined hash positive and indexable
			long offset = (combinedHash & Long.MAX_VALUE) % bitSize;
			if(!jedis.getbit("testKey", offset)){
				return false;
			}
//			if (!bits.get((combinedHash & Long.MAX_VALUE) % bitSize)) {
//				return false;
//			}
			combinedHash += hash2;
		}
		return true;
	}

	/**
	 * 将key存入redis bitmap
	 * @throws Exception 
	 */
	public boolean put(String key) {
		long bitSize = bits.bitSize();
		byte[] bytes = Hashing.murmur3_128()
				.hashObject(key, Funnels.stringFunnel(StandardCharsets.UTF_8))
				.asBytes();
		long hash1 = lowerEight(bytes);
		long hash2 = upperEight(bytes);

		boolean bitsChanged = false;
		long combinedHash = hash1;
		for (int i = 0; i < numHashFunctions; i++) {
			// Make the combined hash positive and indexable
			bitsChanged |= bits.set((combinedHash & Long.MAX_VALUE) % bitSize);
			jedis.setbit("testKey", (combinedHash & Long.MAX_VALUE) % bitSize, true);
			combinedHash += hash2;
		}
		return bitsChanged;
	}

	private /* static */ long lowerEight(byte[] bytes) {
		return Longs.fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3],
				bytes[2], bytes[1], bytes[0]);
	}

	private /* static */ long upperEight(byte[] bytes) {
		return Longs.fromBytes(bytes[15], bytes[14], bytes[13], bytes[12],
				bytes[11], bytes[10], bytes[9], bytes[8]);
	}
}
